package double_stream_join;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.time.Duration;
import java.util.Properties;

public class WindowJoinDemo {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 将全局并行度设为1，方便在同一个控制台查看，当然，可以单独给输出设置并行度为1
        env.setParallelism(1);

        Properties pro = new Properties();
        pro.setProperty("bootstrap.servers", "hadoop102:9092");
        pro.setProperty("group.id", "consumer-group");
        pro.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        pro.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        pro.setProperty("auto.offset.reset", "latest");

        // 从kafka中读取数据
        DataStreamSource<String> kafkaStream = env.addSource(
                new FlinkKafkaConsumer<String>("userbehavior", new SimpleStringSchema(), pro)
        );

        // 将json格式转换成java对象，封装到POJO类里面
        SingleOutputStreamOperator<UserAndBehavior> mapStream = kafkaStream.map(new MapFunction<String, UserAndBehavior>() {
            @Override
            public UserAndBehavior map(String value) throws Exception {
                return FastJsonUtil.getJsonToBean(value, UserAndBehavior.class);
            }
        });

        // 筛选buy行为的数据放入左流orderStream
        SingleOutputStreamOperator<Tuple3<Long, String, Long>> orderStream = mapStream
                .filter(new FilterFunction<UserAndBehavior>() {
                    @Override
                    public boolean filter(UserAndBehavior value) throws Exception {
                        return value.getBehavior().equals("buy");
                    }
                }).map(new MapFunction<UserAndBehavior, Tuple3<Long, String, Long>>() {
                    @Override
                    public Tuple3<Long, String, Long> map(UserAndBehavior value) throws Exception {
                        return Tuple3.of(value.getUser_id(), value.getBehavior(), value.getTs());
                    }
                }).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Tuple3<Long, String, Long>>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<Long, String, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple3<Long, String, Long> element, long recordTimestamp) {
                                        return element.f2;
                                    }
                                })
                );

        // 将pv行为的数据放入右流clickStream
        SingleOutputStreamOperator<UserAndBehavior> clickStream = mapStream.filter(new FilterFunction<UserAndBehavior>() {
            @Override
            public boolean filter(UserAndBehavior value) throws Exception {
                return value.getBehavior().equals("pv");
            }
        }).assignTimestampsAndWatermarks(
                WatermarkStrategy.<UserAndBehavior>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserAndBehavior>() {
                            @Override
                            public long extractTimestamp(UserAndBehavior element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
        );

        orderStream.join(clickStream)
                .where(data -> data.f0) // 指定orderStream流的key
                .equalTo(data -> data.getUser_id()) // 指定clickStream流的key
                .window(TumblingEventTimeWindows.of(Time.seconds(10))) // 滚动窗口10秒
                .apply(new JoinFunction<Tuple3<Long, String, Long>, UserAndBehavior, String>() {
                    @Override
                    public String join(Tuple3<Long, String, Long> first, UserAndBehavior second) throws Exception {
                        return first + "---" + second;
                    }
                }).print();

        env.execute();

    }
}
